package com.heima.kafka.stream;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        KStream<String, String> source = streamsBuilder.stream("stream109_source");
        source
                //数据处理
                //hello kafka  将hello kafka用空格分割成多个数据 [hello ,kafka]
                .flatMapValues((ValueMapper<String, Iterable<String>>) value -> {
                    return Arrays.asList(value.split(" "));
                })
                //分组 hello * 2  kafka * 1 itcast * 1
                .groupBy((key, value) -> value)
                //设置时间窗口，每隔5秒钟处理一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                //统计个数
                .count()
                //将KTable转换成KStream
                .toStream()
                //将key value转换成字符串
                .map((KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>)
                        (objectWindowed, aLong) ->
                                new KeyValue<>(objectWindowed.key(),String.valueOf(aLong)))
                //输入到topic中
                .to("stream109_dest");
        return source;
    }
}